package com.freewater.javabase.demo.mutithread.delayqueue;

import lombok.Data;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.Random;
import java.util.concurrent.*;

/**
 * 参考：<a href="https://zhuanlan.zhihu.com/p/138368078">java中DelayQueue的使用</a>
 */
@Slf4j
public class DelayQueueDemo {

    public static void main(String[] args) {
        DelayQueue<DelayedUser> queue = new DelayQueue<>();
        ExecutorService executor = Executors.newFixedThreadPool(2);
        final long delayTime = 1000;
        final int messageCount = 2;
        executor.submit(new DelayQueueProducer(queue, delayTime, messageCount));
        executor.submit(new DelayQueueConsumer(queue, messageCount));

        try {
            // 禁止新的任务提交，正在运行的任务及排队任务继续执行，直到所有任务执行完毕，线程池随即关闭
            executor.shutdown();
            // 主线程阻塞，等待5秒
            final boolean closed = executor.awaitTermination(5, TimeUnit.SECONDS);
            if (closed) {
                log.info("线程池已关闭");
            } else {
                log.info("线程池关闭超时");
            }
        } catch (InterruptedException e) {
            log.info("主线程 {} interrupted", Thread.currentThread().getName());
        }
    }

    @Data
    public static class DelayedUser implements Delayed {
        private String name;
        private long availableTime;

        public DelayedUser(String name, long delayTime) {
            this.name = name;
            this.availableTime = System.currentTimeMillis() + delayTime;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            long timeDiff = availableTime - System.currentTimeMillis();
            return unit.convert(timeDiff, TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(@NonNull Delayed o) {
            return Long.compare(this.availableTime, ((DelayedUser) o).availableTime);
        }
    }

    public static class DelayQueueProducer extends Thread {
        private final DelayQueue<DelayedUser> queue;
        private final long delayTime;
        private final int messageCount;

        public DelayQueueProducer(DelayQueue<DelayedUser> queue,
                                  long delayTime,
                                  int messageCount) {
            this.queue = queue;
            this.delayTime = delayTime;
            this.messageCount = messageCount;
        }

        @Override
        public void run() {
            final Random random = new Random();
            for (long i = 0; i < messageCount; i++) {
                if (interrupted()) {
                    log.info("DelayQueueProducer interrupted");
                    break;
                }
                final String name = random.nextInt(100) + "";
                DelayedUser user = new DelayedUser(name, this.delayTime);
                queue.add(user);
                log.info("put delayedUser {}", user);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    log.info("DelayQueueProducer interrupted");
                    return;
                }
            }
        }
    }

    public static class DelayQueueConsumer extends Thread {
        private final DelayQueue<DelayedUser> queue;
        private final int messageCount;

        public DelayQueueConsumer(DelayQueue<DelayedUser> queue,
                                  int messageCount) {
            this.queue = queue;
            this.messageCount = messageCount;
        }

        @Override
        public void run() {
            for (int i = 0; i < messageCount; i++) {
                if (interrupted()) {
                    log.info("DelayQueueConsumer interrupted");
                    break;
                }
                try {
                    final DelayedUser user = queue.take();
                    log.info("get delayedUser {}", user);
                } catch (InterruptedException e) {
                    log.info("DelayQueueConsumer interrupted");
                    return;
                }
            }
        }
    }
}


